package cdrprocessing.bolt.util;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import cdrprocessing.common.CallPeriod;

public class ReduceFieldBolt extends BaseRichBolt {

    private SimpleDateFormat df;

    private OutputCollector collector;

    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.df = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
        this.collector = collector;
    }

    public void execute(Tuple input) {
        try {
            final String[] rawCdr = input.getStringByField("rawCdr").split(",");

            if (rawCdr.length == 15) {
                final long customer = Long.valueOf(rawCdr[3]);
                final int location = Integer.valueOf(rawCdr[6]);
                final double duration = Double.valueOf(rawCdr[8]);
                final CallPeriod period = getCallPeriod(rawCdr[9]);
                final int terminateReason = Integer.valueOf(rawCdr[12]);

                collector.emit("CdrStream", new Values(customer, location, duration, period, terminateReason));
            }
        } catch (ParseException e) {
            System.out.println("Invalid tuple");
        }
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declareStream("CdrStream", new Fields("customer", "area", "duration", "period", "terminateReason"));
    }

    private CallPeriod getCallPeriod(String date) throws ParseException {
        int calledTime = df.parse(date).getHours();
        if (calledTime > 0 && calledTime <= 8) {
            return CallPeriod.MORNING;
        } else if (calledTime > 8 && calledTime <= 16) {
            return CallPeriod.AFTERNOON;
        } else {
            return CallPeriod.EVENING;
        }
    }
}
